package com.wande.dataplatform.service.impl;

import cn.hutool.core.util.StrUtil;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.wande.common.core.exception.ServiceException;
import com.wande.common.mybatis.core.page.PageQuery;
import com.wande.common.mybatis.core.page.TableDataInfo;
import com.wande.dataplatform.common.enums.DataPlatformErrorCode;
import com.wande.dataplatform.domain.Job;
import com.wande.dataplatform.domain.JobInstance;
import com.wande.dataplatform.domain.bo.JobInstanceBo;
import com.wande.dataplatform.domain.vo.JobInstanceVO;
import com.wande.dataplatform.mapper.JobInstanceMapper;
import com.wande.dataplatform.mapper.JobMapper;
import com.wande.dataplatform.service.IDolphinSchedulerService;
import com.wande.dataplatform.service.IOpsService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.nio.charset.StandardCharsets;
import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;

/**
 * 运维管理服务实现
 *
 * @author wande
 */
@Slf4j
@Service
@RequiredArgsConstructor
public class OpsServiceImpl implements IOpsService {

    private final JobInstanceMapper jobInstanceMapper;
    private final JobMapper jobMapper;
    private final IDolphinSchedulerService dolphinSchedulerService;

    @Override
    public TableDataInfo<JobInstanceVO> queryInstanceList(JobInstanceBo bo, PageQuery pageQuery) {
        LambdaQueryWrapper<JobInstance> wrapper = buildQueryWrapper(bo);
        Page<JobInstanceVO> page = jobInstanceMapper.selectVoPage(pageQuery.build(), wrapper);
        return TableDataInfo.build(page);
    }

    @Override
    public JobInstanceVO queryInstanceDetail(Long instanceId) {
        JobInstanceVO instance = jobInstanceMapper.selectVoById(instanceId);
        if (instance == null) {
            throw new ServiceException(DataPlatformErrorCode.JOB_INSTANCE_NOT_FOUND.getMessage());
        }
        
        // 如果有DolphinScheduler实例ID，同步最新状态
        if (instance.getDsInstanceId() != null) {
            try {
                dolphinSchedulerService.syncInstanceStatus(instance.getDsInstanceId());
                // 重新查询获取最新状态
                instance = jobInstanceMapper.selectVoById(instanceId);
            } catch (Exception e) {
                log.warn("同步DolphinScheduler实例状态失败: {}", e.getMessage());
            }
        }
        
        return instance;
    }

    @Override
    public JobInstanceVO queryInstanceTree(Long instanceId) {
        JobInstanceVO instance = queryInstanceDetail(instanceId);
        
        // 查询子任务实例（如果有DAG配置）
        if (instance.getDsInstanceId() != null) {
            try {
                List<JobInstanceVO> subInstances = dolphinSchedulerService.getSubTaskInstances(instance.getDsInstanceId());
                instance.setSubInstances(subInstances);
            } catch (Exception e) {
                log.warn("查询子任务实例失败: {}", e.getMessage());
                instance.setSubInstances(new ArrayList<>());
            }
        }
        
        return instance;
    }

    @Override
    public String getInstanceLogs(Long instanceId) {
        JobInstance instance = jobInstanceMapper.selectById(instanceId);
        if (instance == null) {
            throw new ServiceException(DataPlatformErrorCode.JOB_INSTANCE_NOT_FOUND.getMessage());
        }
        
        // 如果有DolphinScheduler实例ID，从DS获取最新日志
        if (instance.getDsInstanceId() != null) {
            try {
                String dsLogs = dolphinSchedulerService.getInstanceLogs(instance.getDsInstanceId());
                if (StrUtil.isNotBlank(dsLogs)) {
                    // 更新本地日志
                    instance.setLogs(dsLogs);
                    instance.setUpdateTime(LocalDateTime.now());
                    jobInstanceMapper.updateById(instance);
                    return dsLogs;
                }
            } catch (Exception e) {
                log.warn("从DolphinScheduler获取日志失败: {}", e.getMessage());
            }
        }
        
        return StrUtil.isNotBlank(instance.getLogs()) ? instance.getLogs() : "暂无日志";
    }

    @Override
    public byte[] downloadInstanceLogs(Long instanceId) {
        String logs = getInstanceLogs(instanceId);
        JobInstance instance = jobInstanceMapper.selectById(instanceId);
        
        // 构建日志文件内容
        StringBuilder logContent = new StringBuilder();
        logContent.append("=".repeat(80)).append("\n");
        logContent.append("作业实例日志\n");
        logContent.append("=".repeat(80)).append("\n");
        logContent.append("实例ID: ").append(instanceId).append("\n");
        logContent.append("作业ID: ").append(instance.getJobId()).append("\n");
        logContent.append("状态: ").append(instance.getStatus()).append("\n");
        logContent.append("开始时间: ").append(instance.getStartTime()).append("\n");
        logContent.append("结束时间: ").append(instance.getEndTime()).append("\n");
        logContent.append("执行时长: ").append(instance.getDuration()).append("ms\n");
        logContent.append("=".repeat(80)).append("\n\n");
        logContent.append(logs);
        
        return logContent.toString().getBytes(StandardCharsets.UTF_8);
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public Long rerunInstance(Long instanceId) {
        JobInstance instance = jobInstanceMapper.selectById(instanceId);
        if (instance == null) {
            throw new ServiceException(DataPlatformErrorCode.JOB_INSTANCE_NOT_FOUND.getMessage());
        }
        
        Job job = jobMapper.selectById(instance.getJobId());
        if (job == null) {
            throw new ServiceException(DataPlatformErrorCode.JOB_NOT_FOUND.getMessage());
        }
        
        // 创建新实例
        JobInstance newInstance = new JobInstance();
        newInstance.setJobId(instance.getJobId());
        newInstance.setStatus("RUNNING");
        newInstance.setStartTime(LocalDateTime.now());
        newInstance.setTriggerType("MANUAL");
        newInstance.setRetryCount(0);
        newInstance.setCreateTime(LocalDateTime.now());
        newInstance.setUpdateTime(LocalDateTime.now());
        
        jobInstanceMapper.insert(newInstance);
        
        // 如果有DolphinScheduler工作流，启动工作流
        if (job.getDsWorkflowId() != null) {
            try {
                Long dsInstanceId = dolphinSchedulerService.startWorkflow(job.getDsWorkflowId());
                newInstance.setDsInstanceId(dsInstanceId);
                jobInstanceMapper.updateById(newInstance);
            } catch (Exception e) {
                log.error("启动DolphinScheduler工作流失败", e);
                newInstance.setStatus("FAILED");
                newInstance.setErrorMessage("启动工作流失败: " + e.getMessage());
                newInstance.setEndTime(LocalDateTime.now());
                jobInstanceMapper.updateById(newInstance);
                throw new ServiceException("启动工作流失败: " + e.getMessage());
            }
        }
        
        log.info("重跑作业实例成功, 原实例ID: {}, 新实例ID: {}", instanceId, newInstance.getId());
        return newInstance.getId();
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public Boolean stopInstance(Long instanceId) {
        JobInstance instance = jobInstanceMapper.selectById(instanceId);
        if (instance == null) {
            throw new ServiceException(DataPlatformErrorCode.JOB_INSTANCE_NOT_FOUND.getMessage());
        }
        
        if (!"RUNNING".equals(instance.getStatus())) {
            throw new ServiceException("只能停止运行中的实例");
        }
        
        // 如果有DolphinScheduler实例，停止DS工作流
        if (instance.getDsInstanceId() != null) {
            try {
                dolphinSchedulerService.stopWorkflow(instance.getDsInstanceId());
            } catch (Exception e) {
                log.error("停止DolphinScheduler工作流失败", e);
                throw new ServiceException("停止工作流失败: " + e.getMessage());
            }
        }
        
        // 更新实例状态
        instance.setStatus("STOPPED");
        instance.setEndTime(LocalDateTime.now());
        instance.setUpdateTime(LocalDateTime.now());
        if (instance.getStartTime() != null) {
            instance.setDuration(java.time.Duration.between(instance.getStartTime(), instance.getEndTime()).toMillis());
        }
        
        jobInstanceMapper.updateById(instance);
        log.info("停止作业实例成功, 实例ID: {}", instanceId);
        return true;
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public Integer batchStopInstances(List<Long> instanceIds) {
        int successCount = 0;
        for (Long instanceId : instanceIds) {
            try {
                stopInstance(instanceId);
                successCount++;
            } catch (Exception e) {
                log.error("停止实例失败, 实例ID: {}, 错误: {}", instanceId, e.getMessage());
            }
        }
        return successCount;
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public Integer batchRerunInstances(List<Long> instanceIds) {
        int successCount = 0;
        for (Long instanceId : instanceIds) {
            try {
                rerunInstance(instanceId);
                successCount++;
            } catch (Exception e) {
                log.error("重跑实例失败, 实例ID: {}, 错误: {}", instanceId, e.getMessage());
            }
        }
        return successCount;
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public Boolean deleteInstance(Long instanceId) {
        JobInstance instance = jobInstanceMapper.selectById(instanceId);
        if (instance == null) {
            throw new ServiceException(DataPlatformErrorCode.JOB_INSTANCE_NOT_FOUND.getMessage());
        }
        
        if ("RUNNING".equals(instance.getStatus())) {
            throw new ServiceException("不能删除运行中的实例，请先停止");
        }
        
        jobInstanceMapper.deleteById(instanceId);
        log.info("删除作业实例成功, 实例ID: {}", instanceId);
        return true;
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public Integer batchDeleteInstances(List<Long> instanceIds) {
        int successCount = 0;
        for (Long instanceId : instanceIds) {
            try {
                deleteInstance(instanceId);
                successCount++;
            } catch (Exception e) {
                log.error("删除实例失败, 实例ID: {}, 错误: {}", instanceId, e.getMessage());
            }
        }
        return successCount;
    }

    /**
     * 构建查询条件
     */
    private LambdaQueryWrapper<JobInstance> buildQueryWrapper(JobInstanceBo bo) {
        LambdaQueryWrapper<JobInstance> wrapper = new LambdaQueryWrapper<>();
        
        wrapper.eq(bo.getJobId() != null, JobInstance::getJobId, bo.getJobId())
            .eq(StrUtil.isNotBlank(bo.getStatus()), JobInstance::getStatus, bo.getStatus())
            .eq(StrUtil.isNotBlank(bo.getTriggerType()), JobInstance::getTriggerType, bo.getTriggerType())
            .ge(bo.getStartTimeBegin() != null, JobInstance::getStartTime, bo.getStartTimeBegin())
            .le(bo.getStartTimeEnd() != null, JobInstance::getStartTime, bo.getStartTimeEnd())
            .orderByDesc(JobInstance::getCreateTime);
        
        return wrapper;
    }
}
